package vip.xiaonuo.bi.core.datasource.util;

import cn.hutool.core.util.StrUtil;
import org.springframework.jdbc.core.JdbcTemplate;
import vip.xiaonuo.bi.core.dataset.enums.DataSetRowType;
import vip.xiaonuo.bi.core.dataset.enums.DorisDataType;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.*;

import static vip.xiaonuo.bi.core.datasource.util.IdUtils.getRandomID;

/**
 * Doris数据库工具类
 *
 *   @author 顾育司
 */
public class DorisUtils {
    private static final String CREATE_TABLE_TEMPLATE = "CREATE TABLE %s (\n%s\n) ";

    private static final String COLUMN_TEMPLATE = "%s %s";

    private static final String UNIQUE_TEMPLATE = " UNIQUE KEY(`%s`)";

    private static final String DISTRIBUTED_TEMPLATE = " DISTRIBUTED BY HASH(`%s`) BUCKETS 1";

    private static final String PROPERTIES_TEMPLATE = " PROPERTIES (\n" +
            "\"replication_allocation\" = \"tag.location.default: 1\",\n" +
            "\"enable_unique_key_merge_on_write\" = \"true\"\n" +
            ")";

    // todo 语句不正确  待补
    public static String generateTable(Map<String, String> columns,String tableName,String uniqueKey) {
        StringBuilder sb = new StringBuilder();
        columns.putIfAbsent(uniqueKey,"VARCHAR(150)");// 给默认主键设置类型
        // 没有的话
        for (Map.Entry<String, String> entry : columns.entrySet()) {
            String columnName = entry.getKey();
            String dorisType = entry.getValue();
            if(uniqueKey.equals(columnName)) {
                // 加入主键第一建立，其他列继续构建表结构
                String columnDefinition = String.format(COLUMN_TEMPLATE, columnName, dorisType);
                sb.append(columnDefinition).append(",\n");
                break;
            }
        }

        for (Map.Entry<String, String> entry : columns.entrySet()) {
            String columnName = entry.getKey();
            String dorisType = entry.getValue();
            if(uniqueKey.equals(columnName)) {
                continue;
            }
            String columnDefinition = String.format(COLUMN_TEMPLATE, columnName, dorisType);
            sb.append(columnDefinition).append(",\n");
        }

        // 删除最后一个逗号和换行符
        if (sb.length() > 0) {
            sb.setLength(sb.length() - 2);
        }
        String createTable = String.format(CREATE_TABLE_TEMPLATE, tableName, sb.toString());
        String  TableSql =createTable +String.format(UNIQUE_TEMPLATE,uniqueKey)
               + String.format(DISTRIBUTED_TEMPLATE,uniqueKey)
               +PROPERTIES_TEMPLATE;
        return TableSql;
    }


    /**
     * 获取数据定义
     * @param tableNameOrColumn
     * @return
     */
    public static String getDefine(String tableNameOrColumn){
        return String.format("`%s`",tableNameOrColumn);

    }

    public static String createTableSql(Map<String, String> Column,String tableName,String partId){

        Map<String, String> DorisColumnTypeMap = new HashMap<>();
        // 替换对应doris类型
        Column.forEach((p,v)->{
            // 替换对应doris类型
            DorisColumnTypeMap.put(p,DorisDataType.getInstance(v).getValue());

        });

        return generateTable(DorisColumnTypeMap,tableName,partId);
    }

    /**
     *
     * @param values  每列的值
     * @param tableName 表名
     * @return
     */
    public static String insertMapData(List<Map<String, String>> values,Map<String, String> columnstype, String tableName, List<String> columns){
        StringBuilder sql = new StringBuilder("INSERT INTO " + tableName + " (");
        for (int i = 0; i < columns.size(); i++) {
            sql.append(columns.get(i));
            if (i < columns.size() - 1) {
                sql.append(",");
            }
        }
        sql.append(") VALUES ");
        for (int i = 0; i < values.size(); i++) {
            Map<String, String> rowValues = values.get(i);
            sql.append("(");
            for (int j = 0; j < columns.size(); j++) {
                String javaType = columnstype.get(columns.get(j));

                String javaValue= rowValues.get(columns.get(j));
                // 列的分类
                DorisDataType type = DorisDataType.getInstance(javaType);

                // 获取可以插入的插入值 如abc 换成 'abc', 123则不变 确保不出错
                String value = DataSetRowType.getType(javaType).converData(javaValue);

                // 分情况组合
                if(value==null){

                    // 数据为空一律插入null
                    sql.append("null");

                }else{
                    // 数值类型不需要''
                    if(type.getType().equals("number")){
                        sql.append( rowValues.get(columns.get(j)) );
                    }else {
                        // 字符型需要''
                        sql.append("'" + rowValues.get(columns.get(j)) + "'");
                    }
                }

                if (j < rowValues.size() - 1) {
                    sql.append(",");
                }
            }
            sql.append(")");
            if (i < values.size() - 1) {
                sql.append(",");
            }
        }
        return sql.append(";").toString();
    }

    /**
     *
     * @param values  每列的值
     * @param tableName 表名
     * @return
     */
    public static String insertListData(List<List<String>> values,Map<String, String> columnstype, String tableName, List<String> columns){
        StringBuilder sql = new StringBuilder("INSERT INTO " + tableName + " (");
        for (int i = 0; i < columns.size(); i++) {
            sql.append(columns.get(i));
            if (i < columns.size() - 1) {
                sql.append(",");
            }
        }
        sql.append(") VALUES ");
        for (int i = 0; i < values.size(); i++) {
            List<String> rowValues = values.get(i);
            sql.append("(");
            for (int j = 0; j < columns.size(); j++) {
                String javaType = columnstype.get(columns.get(j));

                String javaValue= rowValues.get(j);
                // 列的分类
                DorisDataType type = DorisDataType.getInstance(javaType);

                // 获取可以插入的插入值 如abc 换成 'abc', 123则不变 确保不出错
                String value = DataSetRowType.getType(javaType).converData(javaValue);

                // 分情况组合
                if(value==null){

                    // 数据为空一律插入null
                    sql.append("null");

                }else{
                    // 数值类型不需要''
                    if(type.getType().equals("number")){
                        sql.append( javaValue );
                    }else {
                        // 字符型需要''
                        sql.append("'" + javaValue + "'");
                    }
                }

                if (j < rowValues.size() - 1) {
                    sql.append(",");
                }
            }
            sql.append(")");
            if (i < values.size() - 1) {
                sql.append(",");
            }
        }
        return sql.append(";").toString();
    }

    /**
     * 批量插入
     * @param jdbcTemplate
     * @param dataList 数据集
     * @param columnstype  列的类型
     * @param tableName 表名
     * @param columns 列的名称列表
     */
    public static void insertListDataBatch(JdbcTemplate jdbcTemplate,List<List<String>> dataList, Map<String, String> columnstype, String tableName, List<String> columns){

    System.out.println(jdbcTemplate);
    HashMap<String, Object> result = new HashMap<>();
    //结果集中数据为0时,结束方法.进行下一次调用
    if (dataList.size() == 0) {
        return;
    }
    //JDBC分批插入+事务操作完成对10w数据的插入
    Connection conn = null;
    PreparedStatement ps = null;
    try {
        long startTime = System.currentTimeMillis();
        System.out.println(dataList.size() + "条,开始导入到数据库时间:" + startTime + "ms");
        conn = jdbcTemplate.getDataSource().getConnection();
        //控制事务:默认不提交
        conn.setAutoCommit(false);
        String sql = String.format("insert into %s (%s) values",tableName, String.join(",", columns));
        sql += "(?"+ StrUtil.repeat(",?", columns.size()-1)+")"; // (?,?,?,?,?...)

        int batchSize = 0; // 记录当前批处理的数量
        long totalByteSize = 0; // 记录当前批处理的总字节大小
        int flag = 500;  // 500条就计算数据量大小 一次构建（一般不会栈溢出）
        //预编译
        ps = conn.prepareStatement(sql);
        //循环结果集:这里循环不支持"烂布袋"表达式
        for (int i = 0; i < dataList.size(); i++) {
            List<String> rowValues = dataList.get(i);
            for (int j = 0; j < columns.size(); j++) {
                String javaType = columnstype.get(columns.get(j));

                String javaValue= rowValues.get(j);
                // 列的分类
                DorisDataType type = DorisDataType.getInstance(javaType);

                // 获取可以插入的插入值 如abc 换成 'abc', 123则不变 确保不出错
                String value = DataSetRowType.getType(javaType).converData(javaValue);

                // 分情况组合
                if(value==null){
                    // 数据为空一律插入null
                    ps.setString(j+1, null);
                }else{
                    // 数值类型不需要''
                    if(type.getType().equals("number")){
                        ps.setString(j+1,  javaValue );
                    }else {
                        // 字符型需要''
                        ps.setString(j+1,  javaValue );
                    }
                }


            }


            if (flag<500){
                flag ++;
            }

            if (flag>=500){
              int offet =   dataList.size()/500 ;
                // 计算当前值的字节大小并累加到totalByteSize中
//                totalByteSize += getObjectByteSize(rowValues.subList(offet));
                flag  = 0;
            }

            //将一组参数添加到此 PreparedStatement 对象的批处理命令中。
            ps.addBatch();

            batchSize++; // 当前批处理数量加1

            // 判断当前批处理的总字节大小是否超过2G
            if (totalByteSize > 2L * 1024 * 1024 * 1024) {
//            if (totalByteSize > 100) {
                // 执行一次批处理
                ps.executeBatch();
                // 手动提交事务
                conn.commit();
                // 输出当前批处理的数量和总字节大小
                System.out.println("当前批处理数量：" + batchSize + "，总字节大小：" + totalByteSize);
                // 重置batchSize和totalByteSize
                batchSize = 0;
                totalByteSize = 0;
                // 重新创建PreparedStatement对象
                ps = conn.prepareStatement(sql);
            }
        }
        // 执行最后一次批处理,把尾部的也插进去
        ps.executeBatch();
        //手动提交事务
        conn.commit();
        long endTime = System.currentTimeMillis();
        System.out.println(dataList.size() + "条,结束导入到数据库时间:" + endTime + "ms");
        System.out.println(dataList.size() + "条,导入用时:" + (endTime - startTime) + "ms");
        result.put("success", "1111");
    } catch (Exception e) {
        result.put("exception", "0000");
        e.printStackTrace();
    } finally {
        //关毕连接
        try {
            if (ps != null) {
                ps.close();
            }
            if (conn != null) {
                conn.close();
            }
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }
    }
 }

    private static int getObjectByteSize(Object object) {
        try {
            ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
            ObjectOutputStream outputStream = new ObjectOutputStream(byteArrayOutputStream);
            outputStream.writeObject(object);
            outputStream.flush();
            outputStream.close();

            byte[] byteArray = byteArrayOutputStream.toByteArray();
            return byteArray.length;
        } catch (IOException e) {
            e.printStackTrace();
            return -1; // 获取字节大小失败
        }
    }



    /**
     * 获取对应得表的列名
     * @return
     */
    public static Map<String,String> getTableColumn(List<String> columns) {
        Map<String,String> map = new HashMap<>();
        columns.stream().forEach(item->{
            map.put(item,getRandomID());
        });
        return  map;
    }


    public static void main(String[] args) {
        // 1创建表
        Map<String, String> columns = new LinkedHashMap<>();
        columns.put(getDefine("id"), "Long");
        columns.put(getDefine("name"), "String");
        columns.put(getDefine("age"), "INT");
        columns.put(getDefine("salary"), "double");
        columns.put(getDefine("is_active"), "boolean");
        String createTableSql = createTableSql(columns,"`my_table`", "id");
        System.out.println(createTableSql);

        // 2插入语句
        Map<String, String> value1 = new LinkedHashMap<>();
        value1.put("id", "1");
        value1.put("name", "gea");
        value1.put("age", "122");
        Map<String, String> value2 = new LinkedHashMap<>();
        value2.put("id", "2");
        value2.put("name", null);
        value2.put("age", "144");
        List<Map<String, String>> values= new ArrayList<>();
        values.add(value1);
        values.add(value2);
        Map<String, String> type= new HashMap<>();
        type.put("id","double");
        type.put("name","string");
        type.put("age","float");
        ArrayList<String> cols = new ArrayList<>();
        cols.add("id");
        cols.add("name");
        cols.add("age");
        String insertSql =  insertMapData(values,type,"tablename",cols);
        System.out.println(insertSql);
    }
}
